package com.mqtt.service.impl;

import com.alibaba.fastjson.JSON;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.CompletableFuture;

/**
 * Kafka生产者服务
 */
@Slf4j
@Service
public class KafkaProducerService {
    
    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    /**
     * 发送消息到Kafka
     */
    public void sendToKafka(String topic, String key, Object data) {
        try {
            // 构建消息
            Map<String, Object> message = new HashMap<>();
            message.put("key", key);
            message.put("data", data);
            message.put("timestamp", System.currentTimeMillis());
            
            // 异步发送
            ListenableFuture<SendResult<String, Object>> future = 
                kafkaTemplate.send(topic, key, message);
            
            // 添加回调
            future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
                @Override
                public void onSuccess(SendResult<String, Object> result) {
                    log.debug("Message sent to Kafka topic: {}, key: {}, offset: {}", 
                            topic, key, result.getRecordMetadata().offset());
                }
                
                @Override
                public void onFailure(Throwable ex) {
                    log.error("Failed to send message to Kafka topic: {}, key: {}", 
                            topic, key, ex);
                }
            });
            
        } catch (Exception e) {
            log.error("Error sending message to Kafka", e);
        }
    }
    
    /**
     * 同步发送消息到Kafka
     */
    public boolean sendToKafkaSync(String topic, String key, Object data) {
        try {
            Map<String, Object> message = new HashMap<>();
            message.put("key", key);
            message.put("data", data);
            message.put("timestamp", System.currentTimeMillis());
            
            SendResult<String, Object> result = 
                kafkaTemplate.send(topic, key, message).get();
            
            log.debug("Message sent to Kafka topic: {}, partition: {}, offset: {}", 
                    topic, 
                    result.getRecordMetadata().partition(),
                    result.getRecordMetadata().offset());
            
            return true;
            
        } catch (Exception e) {
            log.error("Error sending message to Kafka synchronously", e);
            return false;
        }
    }
    
    /**
     * 批量发送消息
     */
    public CompletableFuture<Boolean> batchSend(String topic, Map<String, Object> messages) {
        return CompletableFuture.supplyAsync(() -> {
            try {
                for (Map.Entry<String, Object> entry : messages.entrySet()) {
                    kafkaTemplate.send(topic, entry.getKey(), entry.getValue());
                }
                log.info("Batch sent {} messages to topic: {}", messages.size(), topic);
                return true;
            } catch (Exception e) {
                log.error("Error in batch send to Kafka", e);
                return false;
            }
        });
    }
    
    /**
     * 发送JSON消息
     */
    public void sendJsonMessage(String topic, String key, Object data) {
        try {
            String jsonData = JSON.toJSONString(data);
            kafkaTemplate.send(topic, key, jsonData);
            log.debug("JSON message sent to Kafka topic: {}", topic);
        } catch (Exception e) {
            log.error("Error sending JSON message to Kafka", e);
        }
    }
}